/*
 *    Copyright 2009 Simone Tripodi
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package com.memos2;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.Future;

import net.spy.memcached.MemcachedClient;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.opensymphony.oscache.base.Config;
import com.opensymphony.oscache.base.persistence.CachePersistenceException;
import com.opensymphony.oscache.base.persistence.PersistenceListener;

/**
 * A memcached based {@link PersistenceListener} for OSCache.
 *
 * @author Simone Tripodi (simone.tripodi)
 * @version $Id: MemcachedListener.java 84 2009-12-15 08:35:31Z simone.tripodi $
 */
@SuppressWarnings("unchecked")
public final class MemcachedListener implements PersistenceListener {

    /**
     * This class log.
     */
    private final Log log = LogFactory.getLog(this.getClass());

    /**
     * The Memcached client configuration.
     */
    private MemcachedConfiguration memcachedConfiguration;

    /**
     * The Memcached client reference.
     */
    private MemcachedClient memcachedClient;

    /**
     * {@inheritDoc}
     */
    public void clear() throws CachePersistenceException {
        // TODO is there a way to clean memcached?!?
    }

    /**
     * {@inheritDoc}
     */
    public PersistenceListener configure(final Config config) {
        this.memcachedConfiguration = MemcachedConfigurationBuilder.getInstance().parseConfiguration(config);

        if (this.log.isDebugEnabled()) {
            this.log.debug("Creating new memcached client using configuration "
                    + this.memcachedConfiguration);
        }

        try {
            this.memcachedClient = new MemcachedClient(this.memcachedConfiguration.getConnectionFactory(), this.memcachedConfiguration.getAddresses());
        } catch (IOException e) {
            String message = "Impossible to instantiate a new memecached client instance, see nested exceptions";
            if (this.log.isFatalEnabled()) {
                this.log.fatal(message, e);
            }
            throw new RuntimeException(message, e);
        }

        return this;
    }

    /**
     * {@inheritDoc}
     */
    public boolean isGroupStored(String groupName) throws CachePersistenceException {
        return this.retrieveGroup(groupName) != null;
    }

    /**
     * {@inheritDoc}
     */
    public boolean isStored(String key) throws CachePersistenceException {
        return this.retrieve(key) != null;
    }

    /**
     * {@inheritDoc}
     */
    public void remove(String key) throws CachePersistenceException {
        String keyString = this.encryptKey(key);

        if (this.log.isDebugEnabled()) {
            this.log.debug("Flushing key '"
                    + keyString
                    + "'");
        }

        this.memcachedClient.delete(keyString);
    }

    /**
     * {@inheritDoc}
     */
    public void removeGroup(String groupName) throws CachePersistenceException {
        String groupKey = this.encryptGroupKey(groupName);

        // remove all group entries
        Set group = (Set) this.retrieveFromMemcached(groupKey);
        for (Object key : group) {
            this.remove((String) key);
        }

        // remove the group entry
        if (this.log.isDebugEnabled()) {
            this.log.debug("Flushing group '"
                    + groupKey
                    + "'");
        }

        this.memcachedClient.delete(groupKey);
    }

    /**
     * {@inheritDoc}
     */
    public Object retrieve(String key) throws CachePersistenceException {
        return this.retrieveFromMemcached(this.encryptKey(key));
    }

    /**
     * {@inheritDoc}
     */
    public Set retrieveGroup(String groupName) throws CachePersistenceException {
        return (Set) this.retrieveFromMemcached(this.encryptGroupKey(groupName));
    }

    /**
     * {@inheritDoc}
     */
    public void store(String key, Object obj) throws CachePersistenceException {
        this.storeInMemcached(this.encryptKey(key), obj);
    }

    /**
     * {@inheritDoc}
     */
    public void storeGroup(String groupName, Set group) throws CachePersistenceException {
        this.storeInMemcached(this.encryptGroupKey(groupName), group);
    }

    /**
     * Retrieves an object from Memcached identified by the passed key.
     *
     * @param key the object unique identifier.
     * @return the identified object if present, null otherwise.
     * @throws CachePersistenceException if any error occurs.
     */
    private final Object retrieveFromMemcached(final String key) throws CachePersistenceException {
        if (this.log.isDebugEnabled()) {
            this.log.debug("Retrieving cached object with key '"
                    + key
                    + "'");
        }

        Object obj = null;

        if (this.memcachedConfiguration.isUsingAsyncGet()) {
            Future<Object> future = this.memcachedClient.asyncGet(key);

            if (this.memcachedConfiguration.isCompressionEnabled()) {
                future = this.memcachedClient.asyncGet(key, new CompressorTranscoder());
            } else {
                future = this.memcachedClient.asyncGet(key);
            }

            try {
                obj = future.get(this.memcachedConfiguration.getTimeout(),
                        this.memcachedConfiguration.getTimeUnit());
            } catch (Exception e) {
                future.cancel(false);
                String errorMessage = "Impossible to retrieve object from memcache identified by '"
                    + key
                    + "' using the async get";

                if (this.log.isErrorEnabled()) {
                    this.log.error(errorMessage, e);
                }

                throw new CachePersistenceException(errorMessage);
            }
        } else {
            if (this.memcachedConfiguration.isCompressionEnabled()) {
                obj = this.memcachedClient.get(key, new CompressorTranscoder());
            } else {
                obj = this.memcachedClient.get(key);
            }
        }

        return obj;
    }

    /**
     * Stores and object identified by the key into Memcached.
     *
     * @param key the object identifier.
     * @param obj the object has to be stored.
     */
    private void storeInMemcached(final String key, final Object obj) {
        if (this.log.isDebugEnabled()) {
            this.log.debug("Storing object ('"
                    + key
                    + "', ="
                    + obj
                    + ")");
        }

        Future<Boolean> future;

        if (this.memcachedConfiguration.isCompressionEnabled()) {
            future = this.memcachedClient.set(key,
                    this.memcachedConfiguration.getExpiration(),
                    obj,
                    new CompressorTranscoder());
        } else {
            future = this.memcachedClient.set(key,
                    this.memcachedConfiguration.getExpiration(),
                    obj);
        }

        if (this.log.isDebugEnabled()) {
            try {
                boolean status = future.get(this.memcachedConfiguration.getTimeout(),
                        this.memcachedConfiguration.getTimeUnit());

                this.log.debug("Status after set object identified by '"
                            + key
                            + "' is '"
                            + status
                            + "'");
            } catch (Exception e) {
                future.cancel(false);

                if (this.log.isWarnEnabled()) {
                    this.log.warn("Impossible to know status after setting object identified by '"
                            + key
                            + "'", e);
                }
            }
        }
    }

    /**
     * Encrypts a key if the encryption is enabled, the same string otherwise.
     *
     * @param key the key has to be encrypted.
     * @return the encrypted key if the encryption is enabled,
     *         the same string otherwise.
     */
    private String encryptKey(final String key) {
        if (this.memcachedConfiguration.isEncryptKeys()) {
            String encrypted = this.memcachedConfiguration.getKeyPrefix() + Integer.toHexString(key.hashCode());

            if (this.log.isDebugEnabled()) {
                this.log.debug("Original key '"
                        + key
                        + "' encrypted to '"
                        + encrypted
                        + "'");
            }

            return encrypted;
        }

        if (this.log.isDebugEnabled()) {
            this.log.debug("No need to encrypt key '"
                    + key
                    + "', just returning the original");
        }

        return key;
    }

    /**
     * Encrypts a group key if the encryption is enabled,
     * the same string otherwise.
     *
     * @param group the group key has to be encrypted.
     * @return the encrypted key if the encryption is enabled,
     *         the same string otherwise.
     */
    private String encryptGroupKey(final String group) {
        if (this.memcachedConfiguration.isEncryptKeys()) {
            String encrypted = this.memcachedConfiguration.getKeyPrefix()
                    + this.memcachedConfiguration.getGroupPrefix()
                    + Integer.toHexString(group.hashCode());

            if (this.log.isDebugEnabled()) {
                this.log.debug("Original group '"
                        + group
                        + "' encrypted to '"
                        + encrypted
                        + "'");
            }

            return encrypted;
        }

        if (this.log.isDebugEnabled()) {
            this.log.debug("No need to encrypt group '"
                    + group
                    + "', just returning the original");
        }

        return group;
    }

}
